package com.alibaba.nacos.core.remote.grpc;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.grpc.auto.Payload;
import com.alibaba.nacos.api.grpc.auto.RequestGrpc;
import com.alibaba.nacos.api.naming.remote.response.InstanceResponse;
import com.alibaba.nacos.api.remote.request.Request;
import com.alibaba.nacos.api.remote.request.RequestMeta;
import com.alibaba.nacos.api.remote.request.ServerCheckRequest;
import com.alibaba.nacos.api.remote.response.ErrorResponse;
import com.alibaba.nacos.api.remote.response.HealthCheckResponse;
import com.alibaba.nacos.api.remote.response.Response;
import com.alibaba.nacos.api.remote.response.ServerCheckResponse;
import com.alibaba.nacos.common.remote.client.grpc.GrpcUtils;
import com.alibaba.nacos.common.utils.JacksonUtils;
import com.alibaba.nacos.core.remote.*;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Objects;

@Service
@Slf4j
public class GrpcRequestAcceptor
        extends RequestGrpc.RequestImplBase {
    @Autowired
    private RequestHandlerRegistry requestHandlerRegistry;
    @Autowired
    private ConnectionManager connectionManager;

    @Override
    public void request(Payload grpcRequest,
                        StreamObserver<Payload> responseObserver) {
        String type = grpcRequest.getMetadata().getType();

        // 服务端检查
        if (Objects.equals(type, ServerCheckRequest.class.getSimpleName())) {
            sendResponse(new ServerCheckResponse(GrpcServerConstants.CONTEXT_KEY_CONN_ID.get(), true), responseObserver);
            return;
        }

        // 未注册服务类型
        RequestHandler requestHandler = requestHandlerRegistry.getByRequestType(type);
        if (requestHandler == null) {
            log.warn("not found handler type:{}", type);
            sendResponse(ErrorResponse.build(NacosException.NO_HANDLER, "RequestHandle not found"), responseObserver);
            return;
        }

        String connectionId = GrpcServerConstants.CONTEXT_KEY_CONN_ID.get();
        boolean requestValid = connectionManager.checkValid(connectionId);
        if (!requestValid) {
            log.warn("Connection is un registered connectionId:{}", connectionId);
            sendResponse(ErrorResponse.build(NacosException.UN_REGISTER, "Connection is un registered."), responseObserver);
            return;
        }

        Request request = GrpcUtils.parse(grpcRequest);
        if (request == null) {
            sendResponse(ErrorResponse.build(NacosException.BAD_GATEWAY, "Invalid request"), responseObserver);
            return;
        }

        try {
            Connection connection = connectionManager.getConnection(connectionId);
            ConnectionMeta connectionMeta = connection.getMetaInfo();
            connection.freshActiveTime();

            RequestMeta requestMeta = new RequestMeta().setClientIp(connectionMeta.getClientIp())
                    .setConnectId(connectionId)
                    .setClientVersion(connectionMeta.getVersion())
                    .setLabels(connectionMeta.getLabels())
                    .setAbilityTable(connection.getAbilityTable());

            Response response = requestHandler.handleRequest(request, requestMeta);
            sendResponse(response, responseObserver);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void sendResponse(Response response,
                              StreamObserver<Payload> responseObserver) {
        if (!(response instanceof HealthCheckResponse)) {
            log.info("send response type:{} response:{}", response == null ? "null" : response.getClass().getSimpleName(), JacksonUtils.toJson(response));
        }
        Payload payload = GrpcUtils.convert(response);
        responseObserver.onNext(payload);
        responseObserver.onCompleted();
    }
}
